package pku;

import pku.ByteMessage;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

public class MessageStore {
	static final MessageStore store = new MessageStore();
	// 消息存储
	HashMap<String, DataOutputStream> topOut = new HashMap<>();// 写
	HashMap<String, DataInputStream> topIn = new HashMap<>();// 读
	static AtomicInteger pushCount = new AtomicInteger();//没写完的push数量

	public void increasePro() {
		pushCount.incrementAndGet();//自增
	}

	public void flush() throws IOException {
		if (pushCount.decrementAndGet() == 0) {//所有文件close
			for (String key : topOut.keySet()) {
				topOut.get(key).close();	
			}
		}
	}

	// push
	public void push(ByteMessage msg, String topic) throws IOException {
		if (msg == null) {
			return;
		}
		DataOutputStream out;
		synchronized (topOut) {
			if (!topOut.containsKey(topic)) {
				out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream("./data/" + topic, true)));
				topOut.put(topic, out);
			} else {
				out = topOut.get(topic);
			}
		}
		// 写入消息
		byte[] body = null;
		byte gize;
	    short heads=0;
	    short tmp=1;
		if (msg.getBody().length > 1024) {
			body = gzip0(msg.getBody());//压缩body
			gize = 1;
		} else {
			body = msg.getBody();
			gize = 0;
		}
		synchronized (out) {
			for (String key : msg.headers().keySet()) {
				byte i=HeaderGize.header2byte.get(key);
				heads=(short) (heads|(tmp<<(i-1)));
			}   
				out.writeShort(heads);
		   for (byte i=1;i<=16;i++) {
			   String key=HeaderGize.byte2header.get(i);
			   if(msg.headers().containsKey(key)){
			       switch (HeaderGize.headerType.get(key)) {
				   case 1:
					   out.writeLong(msg.headers().getLong(key));
					   break;
				   case 2:
					   out.writeDouble(msg.headers().getDouble(key));
					   break;
				   case 3:
					   out.writeInt(msg.headers().getInt(key));
					   break;
				   case 4:
					   out.writeUTF(msg.headers().getString(key));
					   break;
				   }
			   }
		   }
		   out.writeByte(gize);
		   out.writeShort(body.length);
		   out.write(body);
	    }
    }


	public ByteMessage pull(String queue, String topic) throws IOException {
		String k = queue + " " + topic;
		DataInputStream in;
		if (!topIn.containsKey(k)) {
			try {
				in = new DataInputStream(new BufferedInputStream(new FileInputStream("./data/" + topic)));
			} catch (FileNotFoundException e) {
				// e.printStackTrace();
				return null;
			}
			synchronized (topIn) {
				topIn.put(k, in);
			}
		} else {
			in = topIn.get(k);
		}

		if (in.available() != 0) {
			// 读入消息
			ByteMessage msg = new DefaultMessage();
//			byte headtype;
			short heads = in.readShort();
			byte i=1;
			while(heads!=0) {
				if((heads&1)==1) {
					switch (HeaderGize.byte2type.get(i)) {
					case 1:
						msg.putHeaders(HeaderGize.byte2header.get(i), in.readLong());
						break;
					case 2:
						msg.putHeaders(HeaderGize.byte2header.get(i), in.readDouble());
						break;
					case 3:
						msg.putHeaders(HeaderGize.byte2header.get(i), in.readInt());
						break;
					case 4:
						msg.putHeaders(HeaderGize.byte2header.get(i), in.readUTF());
						break;
					}
				}
				heads=(short) (heads>>1);
				i++;
			}

			msg.putHeaders("Topic", topic);//存topic
			byte isZip = in.readByte();
			int len = in.readShort();
			byte[] body = new byte[len];
			in.read(body);
			if (isZip == 1) {
				msg.setBody(gzip1(body));
			} else {
				msg.setBody(body);
			}
			return msg;
		} else {
			return null;
		}
	}
	//进行压缩
	public static byte[] gzip0(byte[] data) {
		byte[] b = null;
		try {
			ByteArrayOutputStream bos = new ByteArrayOutputStream();
			GZIPOutputStream gzip = new GZIPOutputStream(bos);
			gzip.write(data);
			//gzip.finish();
			gzip.close();
			b = bos.toByteArray();
			bos.close();
		} catch (Exception ex) {
			ex.printStackTrace();
		}
		return b;
	}
	//进行解压缩
	public static byte[] gzip1(byte[] data) {
		byte[] b = null;
		try {
			ByteArrayInputStream bis = new ByteArrayInputStream(data);
			GZIPInputStream gzip = new GZIPInputStream(bis);
			byte[] buf = new byte[1024];
			int num = -1;
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			while ((num = gzip.read(buf, 0, buf.length)) != -1) {
				baos.write(buf, 0, num);
			}
			b = baos.toByteArray();
			//baos.flush();
			baos.close();
			gzip.close();
			bis.close();
		} catch (Exception ex) {
			ex.printStackTrace();
		}
		return b;
	}
}
